package com.crois.barrier.web.plugin.route;

import com.crois.barrier.common.constants.Constants;
import com.crois.barrier.common.enums.PluginEnum;
import com.crois.barrier.common.enums.PluginTypeEnum;
import com.crois.barrier.common.exception.BarrierException;
import com.crois.barrier.common.exception.BarrierExceptionEnum;
import com.crois.barrier.plugin.BarrierPlugin;
import com.crois.barrier.plugin.BarrierPluginChain;
import com.crois.barrier.web.resp.BarrierResponseUtils;
import com.crois.barrier.web.resp.BarrierResponseWrapper;
import com.crois.barrier.web.support.ServerWebExchangeUtils;
import io.netty.channel.ConnectTimeoutException;
import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.HttpMethod;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.buffer.NettyDataBuffer;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.server.reactive.AbstractServerHttpResponse;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.netty.http.client.HttpClient;
import reactor.netty.http.client.HttpClientResponse;
import reactor.retry.Backoff;
import reactor.retry.Retry;

import java.time.Duration;
import java.util.Optional;
import java.util.concurrent.TimeoutException;

/**
 * @author Hou Ze Yu
 * @description nettyRoute 插件
 * @date 2020/7/13
 */
@Slf4j
public class NettyHttpClientRoutingPlugin implements BarrierPlugin {

    private static  final Logger LOGGER = LoggerFactory.getLogger(NettyHttpClientRoutingPlugin.class);

    private final HttpClient httpClient;


    public NettyHttpClientRoutingPlugin(HttpClient httpClient) {
        this.httpClient = httpClient;
    }

    @Override
    public Mono<Void> execute(ServerWebExchange exchange, BarrierPluginChain chain) {
        // 获取请求 url
        String  requestUrl = String.valueOf(exchange.getAttributes().get(ServerWebExchangeUtils.BARRIER_REQUEST_URL_ATTR));
        if (StringUtils.isEmpty(requestUrl)){
            return BarrierResponseUtils.error(exchange, BarrierResponseWrapper.fail(BarrierExceptionEnum.BARRIER_GATEWAY_NOT_FIND_AVAILABLE_URL));
        }
        // 开始时间
        Long startTime = System.currentTimeMillis();
        exchange.getAttributes().put(ServerWebExchangeUtils.START_TIME, startTime);
        HttpMethod httpMethod = HttpMethod.valueOf(exchange.getRequest().getMethodValue());
        HttpHeaders filtered = exchange.getRequest().getHeaders();
        DefaultHttpHeaders httpHeaders =new DefaultHttpHeaders();
        filtered.forEach(httpHeaders::set);
        LOGGER.info("[netty httpclient request url :{}]",requestUrl);
        Flux<HttpClientResponse> responseFlux =  this.httpClient.headers(headers->headers.add(httpHeaders))
                .request(httpMethod).uri(requestUrl).send((httpClientRequest,nettyOutbound) ->
//                        nettyOutbound.options(NettyPipeline.SendOptions::flushOnEach).send(
//                                exchange.getRequest().getBody().map(dataBuffer -> ((NettyDataBuffer) dataBuffer).getNativeBuffer())
                                //nettyOutbound.send(request.getBody().map(this::getByteBuf)))
                { return nettyOutbound.send(exchange.getRequest().getBody().map(dataBuffer -> ((NettyDataBuffer) dataBuffer).getNativeBuffer()))
                ;}).responseConnection( (httpClientResponse, connection) -> {
                              exchange.getAttributes().put(ServerWebExchangeUtils.BARRIER_CLIENT_RESPONSE_ATTR,httpClientResponse);
                              exchange.getAttributes().put(ServerWebExchangeUtils.BARRIER_CLIENT_RESPONSE_CONN_ATTR,connection);
                    ServerHttpResponse response = exchange.getResponse();
                    HttpHeaders headers = new HttpHeaders();
                    httpClientResponse.responseHeaders().forEach(stringStringEntry -> headers.add(stringStringEntry.getKey(),stringStringEntry.getValue()));
                     String contentTypeValue = headers.getFirst(HttpHeaders.CONTENT_TYPE);
                     if (org.springframework.util.StringUtils.hasLength(contentTypeValue)){
                              exchange.getAttributes().put(ServerWebExchangeUtils.BARRIER_RESPONSE_CONTENT_TYPE_ATTR,contentTypeValue);
                     }
                    HttpStatus status =HttpStatus.resolve(httpClientResponse.status().code());
                     if (status!=null){
                         response.setStatusCode(status);
                     }else if (response instanceof AbstractServerHttpResponse){
                         ((AbstractServerHttpResponse) response).setStatusCodeValue(httpClientResponse.status().code());
                     }else {
                         throw new BarrierException("unable to set status code on response:"+httpClientResponse.status().code()+":"+response.getClass(),httpClientResponse.status().code());
                     }
                    headers.forEach((key, value) -> {
                            response.getHeaders().putIfAbsent(key,value);
                    });
                    Long executeTime = (System.currentTimeMillis() - startTime);

                    exchange.getAttributes().put(ServerWebExchangeUtils.BARRIER_REQUEST_EXECUTE_TIME, executeTime);

                    LOGGER.info("[netty request url :{},status:{}]",requestUrl,httpClientResponse.status());

                    return Mono.just(httpClientResponse);

                    }).doOnError(e -> {
                    LOGGER.error("[netty request url :{},e:{}]", requestUrl, e.getMessage());
                });
        long timeout = (long) Optional.ofNullable(exchange.getAttribute(ServerWebExchangeUtils.HTTP_TIME_OUT)).orElse(Constants.DEFAULT_BARRIER_HTTP_TIME_OUT);
        long retryTimes =  (long) Optional.ofNullable(exchange.getAttribute(ServerWebExchangeUtils.RETRY_TIMES)).orElse(Constants.DEFAULT_RETRY_TIMES);
        Duration durationTimeout = Duration.ofMillis(timeout);
        responseFlux=responseFlux.timeout(durationTimeout,
                 Mono.error(new TimeoutException("Response longer than timeout"+durationTimeout) ))
                .retryWhen(Retry.onlyIf(x -> x.exception() instanceof ConnectTimeoutException)
                        .retryMax(retryTimes)
                        .backoff(Backoff.exponential(Duration.ofMillis(200), Duration.ofSeconds(20), 2, true)))
                .onErrorMap(TimeoutException.class,throwable ->
                        new BarrierException(BarrierExceptionEnum.BARRIER_GATEWAY_SERVICE_TIMEOUT));
//        Publisher<Void> publisher =  responseFlux.then(chain.execute(exchange));
       // return retryPlugin.apply(exchange,publisher);
        //new TimeoutException("Response longer than timeout"+durationTimeout)
        return responseFlux.then(chain.execute(exchange));
        //ConnectTimeoutException
    }
    @Override
    public PluginTypeEnum pluginType() {
        return PluginTypeEnum.ROUTE;
    }

    @Override
    public String name() {
        return PluginEnum.NETTY_ROUTE.getName();
    }

    @Override
    public int order() {
        return PluginEnum.NETTY_ROUTE.getCode();
    }

    @Override
    public Boolean skip(ServerWebExchange exchange) {
        return false;
    }
}
